package com.yuyunbo.hbase.autoconfigure;

import com.yuyunbo.hbase.constants.Constants;
import com.yuyunbo.hbase.util.HbaseException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.security.UserGroupInformation;

import javax.annotation.Resource;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.*;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author: yuyunbo-1154363414@qq.com
 * @Date: 2020/5/8 20:03
 * @Description:
 */
@Slf4j
public class HbaseClient {
    private String zookeeperList;
    private String zookeeperClientPort;
    private String zookeeperZnodeParent;
    private Connection connection;
    private String userName;
    private String dbNameSpace;
    private String zkSessionTimeout;
    private String hbaseClientRetriesNumber;
    @Resource(name = "hbaseThreadPool")
    private ThreadPoolExecutor hbaseThreadPool;
    Configuration conf = HBaseConfiguration.create();

    protected HbaseClient(String zookeeperList, String zookeeperClientPort, String zookeeperZnodeParent, String userName,
                          String dbNameSpace, String zkSessionTimeout, String hbaseClientRetriesNumber) {
        this.zookeeperList = zookeeperList;
        this.zookeeperClientPort = zookeeperClientPort;
        this.zookeeperZnodeParent = zookeeperZnodeParent;
        this.userName = userName;
        this.dbNameSpace = dbNameSpace;
        this.zkSessionTimeout = zkSessionTimeout;
        this.hbaseClientRetriesNumber = hbaseClientRetriesNumber;
        getConnectin();
    }

    /**
     * 初始化HBase连接。
     * 需配置ZK列表，ZK端口，ZK目录及用户名。
     */
    void getConnectin() {
        conf.set(HConstants.ZOOKEEPER_QUORUM, zookeeperList);
        conf.set(HConstants.ZOOKEEPER_CLIENT_PORT, zookeeperClientPort);
        conf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, zookeeperZnodeParent);
        conf.set(HConstants.ZK_SESSION_TIMEOUT, zkSessionTimeout);
        conf.set(HConstants.HBASE_CLIENT_RETRIES_NUMBER, hbaseClientRetriesNumber);
        User user = User.create(UserGroupInformation.createRemoteUser(userName));
        try {
            this.connection = ConnectionFactory.createConnection(conf, hbaseThreadPool, user);
        } catch (Exception e) {
            log.error("获取Hbase连接异常", e);
            throw new HbaseException("获取Hbase连接异常");
        }
    }

    private static byte[] getDecode(String value) {
        try {
            return value.getBytes("UTF-8");
        } catch (UnsupportedEncodingException e) {
            log.error("字符串字节数组异常", e);
            throw new HbaseException("字符串字节数组异常");
        }
    }

    private String getNameSpace(String tabname) {
        return dbNameSpace.concat(Constants.COLON).concat(tabname);
    }

    /**
     * 单行put。
     * 如果业务允许，建议通过批量put来提升效率。
     */
    public void singlePut(String tbName, String rowKey, String columnFamily, Map<String, String> value) {
        if (value == null || value.isEmpty()) {
            return;
        }
        try (HTable table = (HTable) connection.getTable(TableName.valueOf(getNameSpace(tbName)))) {
            Put put = new Put(getDecode(rowKey));
            Iterator<Map.Entry<String, String>> iterator = value.entrySet().iterator();
            while (iterator.hasNext()) {
                Map.Entry<String, String> entry = iterator.next();
                put.addColumn(getDecode(columnFamily), getDecode(entry.getKey()), getDecode(entry.getValue()));
            }
            table.put(put);
            table.setAutoFlushTo(true);
        } catch (IOException e) {
            log.error(tbName.concat("单个提交数据异常"), e);
            throw new HbaseException(tbName.concat("单个提交数据异常"));
        }
    }

    /**
     * 批量put。
     */
    public void multiplePut(String tbName, String columnFamily, List<Map<String, String>> rowList) {
        try (HTable hTable = (HTable) connection.getTable(TableName.valueOf(getNameSpace(tbName)))) {
            hTable.setAutoFlush(false, false);
            List<Put> list = new ArrayList<>();
            getPutList(list, columnFamily, rowList);
            hTable.put(list);
        } catch (Exception e) {
            log.error(tbName.concat("批量提交数据异常"), e);
            throw new HbaseException(tbName.concat("批量提交数据异常"));
        }
    }

    /**
     * 批量异步put操作
     * <p>
     */
    public void multiplePutAsync(String tbName, String columnFamily, List<Map<String, String>> rowList) {
        List<Put> list = new ArrayList<>();
        try (BufferedMutator bufferedMutator = connection.getBufferedMutator(TableName.valueOf(getNameSpace(tbName)))) {
            getPutList(list, columnFamily, rowList);
            bufferedMutator.mutate(list);
        } catch (Exception e) {
            log.error(tbName.concat("批量提交数据异常"), e);
            throw new HbaseException(tbName.concat("批量提交数据异常"));
        }
    }

    private void getPutList(List<Put> list, String columnFamily, List<Map<String, String>> rowList) {
        for (Map<String, String> rowVlaue : rowList) {
            Put put = new Put(getDecode(rowVlaue.get(Constants.ROWKEY)));
            Iterator<Map.Entry<String, String>> iterator = rowVlaue.entrySet().iterator();
            while (iterator.hasNext()) {
                Map.Entry<String, String> entry = iterator.next();
                if (!Constants.ROWKEY.equals(entry.getKey())) {
                    Object value = entry.getValue();
                    if (value == null) {
                        value = Constants.EMPTY;
                    }
                    put.addColumn(getDecode(columnFamily), getDecode(entry.getKey()), getDecode(value.toString()));
                }
            }
            list.add(put);
        }
    }

    /**
     * 根据相应的条件统计数据总数
     *
     * @param tbName
     * @param family
     * @param paramMap 查询条件 key是表字段名 value是字段值
     * @return
     */
    public long rowCount(String tbName, String family, Map<String, String> paramMap) {
        AggregationClient ac = new AggregationClient(conf);
        Scan scan = new Scan();
        scan.addFamily(getDecode(family));
        if (paramMap != null && paramMap.size() > 0) {
            List<Filter> filters = new ArrayList<>();
            Iterator<Map.Entry<String, String>> iterator = paramMap.entrySet().iterator();
            while (iterator.hasNext()) {
                Map.Entry<String, String> parmaValue = iterator.next();
                if (StringUtils.isNotEmpty(parmaValue.getKey())) {
                    SingleColumnValueFilter filter = new SingleColumnValueFilter(getDecode(family), getDecode(parmaValue.getKey()), CompareFilter.CompareOp.EQUAL, getDecode(parmaValue.getValue()));
                    filters.add(filter);
                }
                FilterList filterList = new FilterList(filters);
                scan.setFilter(filterList);
            }
        }
        long rowCount;
        try {
            rowCount = ac.rowCount(TableName.valueOf(getNameSpace(tbName)), new LongColumnInterpreter(), scan);
        } catch (Throwable throwable) {
            log.error(tbName.concat("统计表数据行项目异常"), throwable);
            throw new HbaseException(tbName.concat("统计表数据行项目异常异常"));
        }
        return rowCount;
    }

    /**
     * hbase 分页查询List
     *
     * @param tbName
     * @param family
     * @param startKey
     * @param pageSize
     * @param paramMap
     * @return
     */
    public List<Map<String, Object>> queryListByParamForPage(String tbName, String family, String startKey, int pageSize, Map<String, String> paramMap) {
        try (Table table = connection.getTable(TableName.valueOf(getNameSpace(tbName)))) {
            Scan scan = new Scan();
            scan.addFamily(getDecode(family));
            FilterList filterList;
            Filter pageFilter = new PageFilter(pageSize);
            List<Filter> filters = new ArrayList<>();
            if (paramMap != null && paramMap.size() > 0) {
                Iterator<Map.Entry<String, String>> iterator = paramMap.entrySet().iterator();
                while (iterator.hasNext()) {
                    Map.Entry<String, String> parmaValue = iterator.next();
                    if (StringUtils.isNotEmpty(parmaValue.getKey())) {
                        SingleColumnValueFilter filter =
                                new SingleColumnValueFilter(getDecode(family), getDecode(parmaValue.getKey()), CompareFilter.CompareOp.EQUAL, getDecode(parmaValue.getValue()));
                        filters.add(filter);
                    }
                }
            }
            if (CollectionUtils.isEmpty(filters)) {
                filterList = new FilterList();
            } else {
                filterList = new FilterList(filters);
            }
            if (StringUtils.isNotEmpty(startKey)) {
                Filter rowFilter = new RowFilter(CompareFilter.CompareOp.GREATER,
                                                        new BinaryComparator(Bytes.toBytes(startKey)));
                filterList.addFilter(rowFilter);
            }
            filterList.addFilter(pageFilter);
            scan.setFilter(filterList);
            ResultScanner results = table.getScanner(scan);
            List<Map<String, Object>> resultList = new ArrayList<>();
            for (Result result : results) {
                resultList.add(analyzeResult(result));
            }
            return resultList;
        } catch (IOException e) {
            log.error(tbName.concat("分页获取数据异常"), e);
            throw new HbaseException(tbName.concat("分页获取数据异常"));
        }
    }

    /**
     * 单行记录get。
     * 如果业务允许，建议通过批量get来提升效率。
     */
    public Map<String, Object> singleGet(String tbName, String rowKey) {
        try (Table table = connection.getTable(TableName.valueOf(getNameSpace(tbName)))) {
            Get get = new Get(getDecode(rowKey));
            Result result = table.get(get);
            return analyzeResult(result);
        } catch (Exception e) {
            log.error(tbName.concat("单行获取记录数据异常"), e);
            throw new HbaseException(tbName.concat("单行获取记录数据异常"));
        }
    }

    /**
     * 单行指定列get。
     * 只查询需要的列，提升get效率。
     */
    public Map<String, Object> singleGetWithColumn(String tbName, String rowKey, String columnFamily, Set<String> columnNameList) {
        try (Table table = connection.getTable(TableName.valueOf(getNameSpace(tbName)))) {
            Get get = new Get(getDecode(rowKey));
            columnNameList.forEach(name -> {
                get.addColumn(getDecode(columnFamily), getDecode(name));
            });
            Result result = table.get(get);
            return analyzeResult(result);
        } catch (Exception e) {
            log.error(tbName.concat("单行获取指定列数据异常"), e);
            throw new HbaseException(tbName.concat("单行获取指定列数据异常"));
        }
    }

    /**
     * 批量get。
     * 提升get效率。
     */
    public List<Map<String, Object>> multipleGet(String tbName, Set<String> rowKeyList, String columnFamily, Set<String> columnNameList) {
        try (Table table = connection.getTable(TableName.valueOf(getNameSpace(tbName)))) {
            List<Get> list = new ArrayList<>();
            rowKeyList.forEach(row -> {
                Get get = new Get(getDecode(row));
                columnNameList.forEach(name -> {
                    get.addColumn(getDecode(columnFamily), getDecode(name));
                });
                list.add(get);
            });
            Result[] results = table.get(list);
            List<Map<String, Object>> resultList = new ArrayList<>();
            for (Result result : results) {
                resultList.add(analyzeResult(result));
            }
            return resultList;
        } catch (IOException e) {
            log.error(tbName.concat("批量获取指定列数据异常"), e);
            throw new HbaseException(tbName.concat("批量获取指定列数据异常"));
        }
    }

    /**
     * 处理Result数据。
     */
    Map<String, Object> analyzeResult(Result result) {
        Map<String, Object> resultMap = new HashMap<>();
        String rowKey = getStringByByteArray(result.getRow());
        for (Cell c : result.rawCells()) {
            resultMap.put(getStringByByteArray(CellUtil.cloneQualifier(c)), getStringByByteArray(CellUtil.cloneValue(c)));
        }
        resultMap.put("rowKey", rowKey);
        return resultMap;
    }

    String getStringByByteArray(byte[] value) {
        return Bytes.toString(value);
    }

    /**
     * 扫描数据。
     * 适用于查询多条符合条件的数据。
     * setStartRow(ROWKEY) ：设置start key
     * setStopRow(ROWKEY_2) ：设置end key
     * setCaching(100) : 设置单次rpc（最大）查询的数量。增大可以减少rpc次数，但是可能会导致timeout。减小会增加rpc次数，影响效率。
     */
    public List<Map<String, Object>> scanSingleColumnValueFilter(String tbName, String startRow, String stopRow, String columnFamily, String columnName, String value) {
        try (Table table = connection.getTable(TableName.valueOf(getNameSpace(tbName)))) {
            Scan scan = new Scan();
            scan.withStartRow(getDecode(startRow));
            scan.withStopRow(getDecode(stopRow));
            scan.setCaching(100);
            SingleColumnValueFilter filter =
                    new SingleColumnValueFilter(getDecode(columnFamily), getDecode(columnName), CompareFilter.CompareOp.EQUAL, getDecode(value));
            scan.setFilter(filter);
            // 设置过滤器，视查询场景而定。
            ResultScanner scanner = table.getScanner(scan);
            return analyzeScanner(scanner);
        } catch (IOException e) {
            log.error(tbName.concat("扫描获取指定列数据异常"), e);
            throw new HbaseException(tbName.concat("扫描获取指定列数据异常"));
        }
    }

    /**
     * 根据key的前缀扫描数据。
     * 不建议使用RowFilter或者PrefixFilter,可能造成全表扫描，性能极差。
     */
    public List<Map<String, Object>> scanByPrefixFilter(String tbName, String prefix) {
        try (Table table = connection.getTable(TableName.valueOf(getNameSpace(tbName)))) {
            Scan scan = new Scan();
            byte[] startRow = Bytes.toBytes(prefix);
            scan.withStartRow(startRow);
            scan.withStopRow(Bytes.add(startRow, new byte[]{Byte.MAX_VALUE}));
            ResultScanner scanner = table.getScanner(scan);
            return analyzeScanner(scanner);
        } catch (IOException e) {
            log.error(tbName.concat("扫描指定前缀获取指定列数据异常"), e);
            throw new HbaseException(tbName.concat("扫描指定前缀获取指定列数据异常"));
        }
    }

    /**
     * 处理Scanner数据。使用万Scanner后，必须关闭Scanner！
     */
    List<Map<String, Object>> analyzeScanner(ResultScanner scanner) {
        try {
            List<Map<String, Object>> resultList = new ArrayList<>();
            for (Result result : scanner) {
                resultList.add(analyzeResult(result));
            }
            return resultList;
        } finally {
            if (scanner != null) {
                scanner.close();
            }
        }
    }
}
